package com.sunzm.flink.datastream.scala.robotcall

import com.alibaba.fastjson.JSON
import org.apache.commons.lang3.StringUtils
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

/**
 *
 * 费率变化的数据处理函数
 *
 * @author Administrator
 * @version 1.0
 * @date 2021-08-05 21:54
 */
class RobotCallFeeRateChangeProcessFunction extends ProcessFunction[String, String] {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName.stripSuffix("$"))

  override def processElement(line: String,
                              ctx: ProcessFunction[String, String]#Context,
                              out: Collector[String]): Unit = {

    try {
      val jSONObject = JSON.parseObject(line)

      if (jSONObject.containsKey("data") && jSONObject.containsKey("type")) {
        //操作类型
        val opType = StringUtils.trim(jSONObject.getString("type")).toUpperCase

        val dataJSONObject = jSONObject.getJSONObject("data")
        val number_type = dataJSONObject.getIntValue("number_type")
        val sip_callin_charge = dataJSONObject.getDoubleValue("sip_callin_charge")
        val sip_callout_charge = dataJSONObject.getDoubleValue("sip_callout_charge")
        val phone_callin_charge = dataJSONObject.getDoubleValue("phone_callin_charge")
        val phone_callout_charge = dataJSONObject.getDoubleValue("phone_callout_charge")

        jSONObject.clear()
        jSONObject.put("number_type", number_type)
        jSONObject.put("sip_callin_charge", sip_callin_charge)
        jSONObject.put("sip_callout_charge", sip_callout_charge)
        jSONObject.put("phone_callin_charge", phone_callin_charge)
        jSONObject.put("phone_callout_charge", phone_callout_charge)
        jSONObject.put("opType", opType)

        out.collect(jSONObject.toJSONString)
      }
    } catch {
      case e: Throwable => {
        logger.warn(s"JSON解析异常: ${line}", e)
      }
    }
  }
}